package com.mq.rabbitmq.point;

import com.mq.rabbitmq.ConsumeZone5;
import com.mq.rabbitmq.deadletter.DealLetterConfig;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.Primary;
import org.springframework.context.annotation.Scope;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.interceptor.RetryOperationsInterceptor;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

import java.util.concurrent.Executor;


/**
 * Direct模式,可以使用rabbitMQ自带的Exchange：default Exchange 。所以不需要将Exchange进行任何绑定(binding)操作 。消息传递时，RouteKey必须完全匹配，才会被队列接收，否则该消息会被抛弃。
 * 创建的对列，默认都会和default Exchange 绑定，routihgkey使用自己的名称
 * 当你手动创建一个队列时，后台会自动将这个队列绑定到一个名称为空的Direct类型交换机上，绑定路由名称与队列名称相同
 * Created by xi on 2017/8/1.
 */
@EnableRabbit
@Import(value = {ConsumeZone5.class,Send.class,PointListener.class})
public class AmqpConfig {
    public static final String EXCHANGE   = "spring-boot-exchange";
    public static final String ROUTINGKEY = "spring-boot-routingKey-xcw";

    public static final String  NO_BIND_QUEQU = "xiaocai-no-bind-queue";
    public static final String  QUEUE_DEAD ="queue.dlx.test";

    @Bean
    public Queue testQueue232() {
        Queue ue =  new Queue(AmqpConfig.ROUTINGKEY); //队列持久
        return ue;
    }

    //    @Bean
//    public Queue testQueue2() {
//        Queue ue =  new Queue("x2"); //队列持久
//        return ue;
//    }
//    @Bean
//    public Queue testQueue3() {
//        Queue ue =  new Queue("x3"); //队列持久
//        return ue;
//    }
//
//    @Bean
//    public Queue testQueu4() {
//        Queue ue =  new Queue("x4"); //队列持久
//        return ue;
//    }
//    @Bean
//    public Queue testQueue5() {
//        Queue ue =  new Queue("x5"); //队列持久
//        return ue;
//    }
//    @Bean
//    public Queue testQueue6() {
//        Queue ue =  new Queue("x6"); //队列持久
//        return ue;
//    }
//
//    @Bean
//    public Queue testQueue7() {
//        Queue ue =  new Queue("x7"); //队列持久
//        return ue;
//    }
//    @Bean
//    public Queue testQueue8() {
//        Queue ue =  new Queue("x8"); //队列持久
//        return ue;
//    }
//
//
//    @Bean
//    public Queue testQueue9() {
//        Queue ue =  new Queue("x9"); //队列持久
//        return ue;
//    }
    @Bean
    public Queue testQueue10() {
        Queue ue =  new Queue("x10"); //队列持久
        return ue;
    }

    /**
     *
     * 正常情况下，Spring Boot会在首次发送消息时，自动创建所需要的Exchange和Queue
     *
     * 针对消费者配置
     * 1. 设置交换机类型
     * 2. 将队列绑定到交换机
     *
     *
     FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念
     HeadersExchange ：通过添加属性key-value匹配
     DirectExchange:按照routingkey分发到指定队列
     TopicExchange:多关键字匹配
     */
    @Bean
    public DirectExchange defaultExchange() {
        return new DirectExchange(AmqpConfig.EXCHANGE);
    }

    /**
     * 其他地方也有这个bean 的Id,这个也不能走了,不能直接叫binding
     * @return
     */
    @Bean
    public Binding testQueueQinding232(){
        return BindingBuilder.bind(testQueue232()).to(new DirectExchange(AmqpConfig.EXCHANGE)).with(testQueue232().getName());
    }
    @Bean
    public Executor defaultThreadPool() {
        ThreadPoolTaskScheduler executor = new ThreadPoolTaskScheduler();
        executor.setPoolSize(1);
        executor.setThreadPriority(1);
        executor.setThreadNamePrefix("testshiro_");
        executor.initialize();
        return executor;
    }

    /**
     * 条件：机器2是内存队列，机器1 是磁盘队列，没有设置同步镜像队列模式
     * 场景1：机器2设置一个队列数据，然后机器2挂掉，机器1去连的时候，队列不存在，就报错了。队列挂的时候，就报错了
     * home node 'rabbit@lc-dev-rabbitmq2' of durable queue 'test-break-onenode3' in vhost '/' is down or inaccessible, class-id=50, method-id=10)
     *
     * 场景2：机器2重启之后，看持久化数据是否是正常的吗？
     * 测试发现
     *
     *
     *
     * @return
     */

//    /**
//     * 没有这个，自动创建就失败了，需要加入这个bean
//     * @param connectionFactory
//     * @return
//     */
    @Bean
    public RabbitAdmin rabbitAdmin(){
        return new RabbitAdmin(connectionFactory());
    }

    /**
     * 在使用RabbitMQ的时候，我们可以通过消息持久化操作来解决因为服务器的异常奔溃导致的消息丢失，除此之外我们还会遇到一个问题，当消息的发布者在将消息发送出去之后，消息到底有没有正确到达broker代理服务器呢？如果不进行特殊配置的话，默认情况下发布操作是不会返回任何信息给生产者的，
     * 也就是默认情况下我们的生产者是不知道消息有没有正确到达broker的，如果在消息到达broker之前已经丢失的话，持久化操作也解决不了这个问题，因为消息根本就没到达代理服务器，你怎么进行持久化，那么这个问题该怎么解决呢？
     * @return
     */
    @Bean(name="firstConnectionFactory")
    @Primary
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
//        connectionFactory.setAddresses("10.10.20.55:5672");
//        connectionFactory.setUsername("qfrqprod");
//        connectionFactory.setPassword("qfrqprod7089");

//        connectionFactory.setAddresses("192.168.2.47:5672");
//        connectionFactory.setUsername("admin");
//        connectionFactory.setPassword("rabbitmqadmin");

        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setChannelCacheSize(100);
        connectionFactory.setChannelCheckoutTimeout(200);

        //服务器发送给正常的消费者成功了，会回调，不然的话，生产者不知道是否服务器正常推送给消费者了。
        //消费者是否正常消费了，假如持久化失败，
        //消息服务器异常，导致该消息丢失
        connectionFactory.setPublisherConfirms(true); //必须要设置
        return connectionFactory;
    }

    /**
     *消费者重试模版
     * @return
     */
    @Bean
    RetryOperationsInterceptor interceptor() {
        //设置backoff 策略 重试时间间隔策略
        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(1000); // 执行完了，过1s继续执行
        backOffPolicy.setMultiplier(2);    //乘数 1,2,4,8,16 s执行
        backOffPolicy.setMaxInterval(15000);

        return RetryInterceptorBuilder.stateless()
                .maxAttempts(5)
                .backOffPolicy(backOffPolicy)
                        //.recoverer(new RejectAndDontRequeueRecoverer())
                .recoverer(new RepublishMessageRecoverer(rabbitTemplate(), DealLetterConfig.DEFAULT_DEAD_EXCHANGE, DealLetterConfig.DEAD_QUEUE))
                .build();
    }


    /**
     * 发送者重试模版
     * @return
     */
    @Bean
    public RetryTemplate retryTemplate(){
        RetryTemplate retryTemplate = new RetryTemplate();
        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(1000);
        backOffPolicy.setMaxInterval(15000);
        backOffPolicy.setMultiplier(2);
        retryTemplate.setBackOffPolicy(backOffPolicy);

        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(5);

        retryTemplate.setRetryPolicy(retryPolicy);
        return retryTemplate;
    }



    //必须是prototype类型
    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        template.setMessageConverter(new Jackson2JsonMessageConverter());
        //Create retries configuration for message delivery
       // template.setRetryTemplate(retryTemplate());
        return template;
    }

}